package com.sunday.common.mq.rabbit.study.base.e20_IntroducingPublisherConfirms;

import com.rabbitmq.client.*;
import com.sunday.common.mq.rabbit.study.base.ConnectionUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * https://blog.rabbitmq.com/posts/2011/02/introducing-publisher-confirms
 * <p>
 * 介绍发布者确认
 * 在许多消息传递方案中，不得丢失消息。
 * 由于 AMQP 对消息持久性/处理几乎没有保证，因此执行此操作的传统方法是使用事务，这可能慢得令人无法接受。为了解决这个问题，我们以轻量级发布者确认的形式引入了对 AMQP 的扩展。
 */
@Slf4j
public class IntroducingPublisherConfirms {


    /**
     * 保证交付与TX一起
     * 在 RabbitMQ 中，持久消息是在代理重新启动后应该在后继续存在的消息。
     * 这里的关键词是应该，因为如果代理在有机会将消息写入磁盘之前关闭，消息仍然会丢失。
     * 在某些情况下，这还不够，发布者需要知道消息是否得到了正确处理。直接的解决方案是使用事务，即提交每条消息。
     */
    @Test
    @SneakyThrows
    public void tx() {

        Random rand = new Random();

        ConnectionFactory cf = ConnectionUtil.factory();
        try (Connection connection = cf.newConnection()) {
            Channel ch = connection.createChannel();

            String queue = UUID.randomUUID().toString();
            ch.queueDeclare(queue, false, false, true, null);
            // 在此通道上启用发布者确认。
            ch.txSelect();
            for (int i = 0; i < 10; i++) {
                Integer value = rand.nextInt(5);
                ch.basicPublish("", queue, null, String.valueOf(value).getBytes());
                if (value > 3) {
                    log.info("[success] -> value : {}", value);
                    ch.txCommit();
                } else {
                    log.info("[fail] -> value : {}", value);
                    ch.txRollback();
                }
            }

            ch.basicConsume(queue, false, (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
                log.info(" consumerTag : {} ", consumerTag);
                log.info(" envelope : {} ", delivery.getEnvelope());
                log.info(" properties : {} ", delivery.getProperties());
                log.info(" [x] Received '{}'", message);
            }, consumerTag -> {
            });

            TimeUnit.SECONDS.sleep(10);

        }
    }

}
